// Copyright (C) Calum Grant 2008

#include "cppscript"
#include "dynamic/extensions.hpp"
#include "thread.hpp"


namespace
{
	namespace Queue
	{
		var pop(var queue)
		{
			{
				internal::leave_apartment l;
				queue["Queue::num_waiting"].as<api::stack>().wait_pop();
			}
			return queue["Queue::pending"].pop_front();
		}

		void post(var queue, var message)
		{
			queue["Queue::pending"].push_back(message);
			queue["Queue::num_waiting"].as<api::stack>().push();
		}

		void wait(var queue)
		{
			var num_waiting = queue["Queue::num_waiting"];
			internal::leave_apartment l;
			num_waiting.as<api::stack>().wait_empty();
		}
	}


	namespace MessageQueue
	{
		void post(var queue, var message)
		{
			Queue::post(queue, message);
			queue["MessageQueue::num_done"].as<api::stack>().push();
		}

		void process(var queue)
		{
			var item = queue["pop"]();

			try
			{
				queue["consumer"](item);
			}
			catch(...)
			{
				// Silently ignore exceptions (not good design)
			}

			internal::leave_apartment l;
			queue["MessageQueue::num_done"].as<api::stack>().wait_pop();
		}

		void process_thread(var queue)
		{
			while( !queue["Queue::num_waiting"].as<api::stack>().check_shutdown() )
			{
				process(queue);
			}
		}

		void close(var queue)
		{
			queue["Queue::num_waiting"].as<api::stack>().shutdown();
			queue["MessageQueue::num_done"].as<api::stack>().shutdown();

			foreach(thread, queue["threads"])
				thread["join"]();
		}

		void wait(var queue)
		{
			Queue::wait(queue);
			var num_done = queue["MessageQueue::num_done"];
			internal::leave_apartment l;
			num_done.as<api::stack>().wait_empty();
		}
	}
}


var dynamic::queue()
{
	return object("Queue").extend
		("Queue::pending", list())
		("post", Queue::post)
		("Queue::num_waiting", create<api::stack>())
		("pop", Queue::pop)
		("wait", Queue::wait);
}


var dynamic::message_queue(const var & consumer, const var & num_threads)
{
	var q = queue().extend
		("process_thread", MessageQueue::process_thread)
		("consumer", consumer)
		("process", MessageQueue::process)
		("threads", array())
		("close", MessageQueue::close)
		("post", MessageQueue::post)
		("MessageQueue::num_done", create<api::stack>())
		("wait", MessageQueue::wait);

	for(var i = 0; i<num_threads; ++i)
	{
		q["threads"].push_back(thread(q["process_thread"]));
	}

	return q;
}
